package com.binance.actuator.indicator;

import java.util.Map;
import java.util.Properties;

import org.redisson.spring.data.connection.RedissonConnectionFactory;
import org.springframework.boot.actuate.health.AbstractHealthIndicator;
import org.springframework.boot.actuate.health.CompositeHealthIndicator;
import org.springframework.boot.actuate.health.Health;
import org.springframework.boot.actuate.health.HealthAggregator;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.data.redis.connection.ClusterInfo;
import org.springframework.data.redis.connection.RedisClusterConnection;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisConnectionUtils;
import org.springframework.util.Assert;

import lombok.extern.log4j.Log4j2;

/**
 * Created by robin.wu on 2018/11/14.
 */
@Log4j2
public class RedisHealthIndicator extends AbstractHealthIndicator {
    private static final String VERSION = "version";

    private static final String REDIS_VERSION = "redis_version";

    private final RedisConnectionFactory redisConnectionFactory;

    public RedisHealthIndicator(RedisConnectionFactory connectionFactory) {
        Assert.notNull(connectionFactory, "ConnectionFactory must not be null");
        this.redisConnectionFactory = connectionFactory;
    }

    @Override
    protected void doHealthCheck(Health.Builder builder) throws Exception {
        RedisConnection connection;
        if(redisConnectionFactory instanceof RedissonConnectionFactory) {
            try {
                connection = redisConnectionFactory.getClusterConnection();
            }catch (Exception e) {
                // redisson connection do not support info ,but cluster connection does
                log.error("redis is not cluster mode");
                builder.up();
                return;
            }
        }else {
            connection = RedisConnectionUtils.getConnection(this.redisConnectionFactory);
        }

        try {
            if (connection instanceof RedisClusterConnection) {
                ClusterInfo clusterInfo = ((RedisClusterConnection) connection)
                        .clusterGetClusterInfo();
                builder.up().withDetail("cluster_size", clusterInfo.getClusterSize())
                        .withDetail("slots_up", clusterInfo.getSlotsOk())
                        .withDetail("slots_fail", clusterInfo.getSlotsFail());
            }
            else {
                Properties info = connection.info();
                builder.up().withDetail(VERSION, info.getProperty(REDIS_VERSION));
            }
        }
        finally {
            RedisConnectionUtils.releaseConnection(connection,this.redisConnectionFactory);
        }
    }


    public static HealthIndicator createHealthIndicator(Map<String, RedisConnectionFactory> beans, HealthAggregator healthAggregator) {
        if (beans.size() == 1) {
            return createHealthIndicator(beans.values().iterator().next());
        }
        CompositeHealthIndicator composite = new CompositeHealthIndicator(healthAggregator);
        for (Map.Entry<String, RedisConnectionFactory> entry : beans.entrySet()) {
            composite.addHealthIndicator(entry.getKey(),createHealthIndicator(entry.getValue()));
        }
        return composite;
    }

    public static RedisHealthIndicator createHealthIndicator(RedisConnectionFactory redisConnectionFactory) {
        return new RedisHealthIndicator(redisConnectionFactory);
    }
}
